package com.yupi.springbootinit.repository;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import com.influxdb.client.QueryApi;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Repository;

import java.time.Instant;
import java.util.*;
import java.util.stream.Collectors;


/**
 * InfluxDB 2.x 仓储层（Flux 查询）
 */
@Repository
@Slf4j
public class InfluxRepository {

    @Value("${influx.url}")
    private String influxUrl;

    @Value("${influx.token}")
    private String influxToken;

    @Value("${influx.org}")
    private String orgName;

    @Value("${influx.bucket}")
    private String bucket;

    @Value("${influx.latestRecentRange:-30m}")
    private String latestRecentRange; // 近窗口，优先使用

    @Value("${influx.latestFallbackRange:-30d}")
    private String latestFallbackRange; // 回退窗口，用于补齐缺失字段

    @Value("${influx.aggWindow:4m}")
    private String aggWindow;

    @Value("${influx.aggFn:mean}")
    private String aggFn;

    private static final String MEASUREMENT = "modbus_data";

    private InfluxDBClient client;
    private QueryApi queryApi;

    @PostConstruct
    public void init() {
        client = InfluxDBClientFactory.create(influxUrl, influxToken.toCharArray(), orgName, bucket);
        queryApi = client.getQueryApi();
    }

    @PreDestroy
    public void cleanup() {
        if (client != null) {
            client.close();
        }
    }

    // ========== 最新值（单设备，按 IP）==========
    public Optional<Map<String, Object>> queryLatestByIp(String ip) {
        String ipEsc = escapeTag(ip);
        String flux = String.format("""
                        recent = from(bucket: "%s")
                          |> range(start: %s)
                          |> filter(fn: (r) => r._measurement == "%s" and r.ip == "%s")
                          |> group(columns: ["_field"])
                          |> last()
                          |> keep(columns: ["_field","_value","_time"])
                        
                        fallback = from(bucket: "%s")
                          |> range(start: %s, stop: now())
                          |> filter(fn: (r) => r._measurement == "%s" and r.ip == "%s")
                          |> group(columns: ["_field"])
                          |> last()
                          |> keep(columns: ["_field","_value","_time"])
                        
                        union(tables: [recent, fallback])
                          |> group(columns: ["_field"])
                          |> last()
                        """,
                bucket, latestRecentRange, MEASUREMENT, ipEsc,
                bucket, latestFallbackRange, MEASUREMENT, ipEsc
        );
        return queryLatestAsMap(flux);
    }

    // ========== 最新值（单设备，按 name）==========
    public Optional<Map<String, Object>> queryLatestByName(String measurement, String tagName) {
        String m = escapeIdent(measurement);
        String n = escapeTag(tagName);
        String flux = String.format("""
                        recent = from(bucket: "%s")
                          |> range(start: %s)
                          |> filter(fn: (r) => r._measurement == "%s" and r.name == "%s")
                          |> group(columns: ["_field"])
                          |> last()
                          |> keep(columns: ["_field","_value","_time"])
                        
                        fallback = from(bucket: "%s")
                          |> range(start: %s, stop: now())
                          |> filter(fn: (r) => r._measurement == "%s" and r.name == "%s")
                          |> group(columns: ["_field"])
                          |> last()
                          |> keep(columns: ["_field","_value","_time"])
                        
                        union(tables: [recent, fallback])
                          |> group(columns: ["_field"])
                          |> last()
                        """,
                bucket, latestRecentRange, m, n,
                bucket, latestFallbackRange, m, n
        );
        return queryLatestAsMap(flux);
    }

    private Optional<Map<String, Object>> queryLatestAsMap(String flux) {
        List<FluxTable> tables = queryApi.query(flux);
        if (tables == null || tables.isEmpty()) return Optional.empty();

        Map<String, Object> result = new HashMap<>();
        Instant latestTime = null;

        for (FluxTable table : tables) {
            for (FluxRecord rec : table.getRecords()) {
                Instant t = rec.getTime() != null ? rec.getTime() : null;
                if (t != null && (latestTime == null || t.isAfter(latestTime))) {
                    latestTime = t;
                }
                if (rec.getField() != null) {
                    result.put(rec.getField(), rec.getValue());
                }
            }
        }
        if (latestTime != null) {
            result.put("time", latestTime.toString());
        }
        return result.isEmpty() ? Optional.empty() : Optional.of(result);
    }

    // ========== 历史：原始 ==========
    public List<HistoryPoint> queryHistoryRawByName(String measurement, String tagName, List<String> fields, String start) {
        String m = escapeIdent(measurement);
        String n = escapeTag(tagName);
        String fieldFilter = fields.stream()
                .map(f -> String.format("r._field == \"%s\"", escapeTag(f)))
                .collect(Collectors.joining(" or "));

        String flux = String.format(
                "from(bucket: \"%s\") "
                        + "|> range(start: %s) "
                        + "|> filter(fn: (r) => r._measurement == \"%s\" and r.name == \"%s\") "
                        + "|> filter(fn: (r) => %s) "
                        + "|> yield(name: \"raw\")",
                bucket, start, m, n, fieldFilter
        );

        return queryPoints(flux);
    }

    // ========== 历史：聚合 ==========
    public List<HistoryPoint> queryHistoryAggByName(String measurement, String tagName, List<String> fields, String start, String agg) {
        String m = escapeIdent(measurement);
        String n = escapeTag(tagName);
        String fieldFilter = fields.stream()
                .map(f -> String.format("r._field == \"%s\"", escapeTag(f)))
                .collect(Collectors.joining(" or "));

        String flux = String.format(
                "from(bucket: \"%s\") "
                        + "|> range(start: %s) "
                        + "|> filter(fn: (r) => r._measurement == \"%s\" and r.name == \"%s\") "
                        + "|> filter(fn: (r) => %s) "
                        + "|> aggregateWindow(every: %s, fn: mean, createEmpty: false) "
                        + "|> yield(name: \"agg\")",
                bucket, start, m, n, fieldFilter, agg
        );

        return queryPoints(flux);
    }

    private List<HistoryPoint> queryPoints(String flux) {
        List<FluxTable> tables = queryApi.query(flux);
        List<HistoryPoint> list = new ArrayList<>();
        if (tables == null) return list;
        for (FluxTable table : tables) {
            for (FluxRecord r : table.getRecords()) {
                HistoryPoint p = new HistoryPoint();
                p.time = r.getTime() != null ? r.getTime().toString() : null;
                p.field = r.getField();
                Object v = r.getValue();
                p.value = v instanceof Number ? ((Number) v).doubleValue() : null;
                list.add(p);
            }
        }
        return list;
    }

    public static class HistoryPoint {
        public String time;
        public String field;
        public Double value;
    }

    // ========== 状态序列 ==========
    public List<HistoryPoint> queryStatusSeries(String measurement, String tagName, String start, String end, String every) {
        String m = escapeIdent(measurement);
        String n = escapeTag(tagName);
        String flux = String.format(
                "from(bucket: \"%s\") "
                        + "|> range(start: %s, stop: %s) "
                        + "|> filter(fn: (r) => r._measurement == \"%s\" and r.name == \"%s\" and r._field == \"statusCode\") "
                        + "|> aggregateWindow(every: %s, fn: mean, createEmpty: false) "
                        + "|> yield(name: \"status\")",
                bucket, start, end, m, n, every
        );
        return queryPoints(flux);
    }

    // ========== 首次 printStart=1 的时间 ==========
    public String queryFirstPrintStartTime(String measurement, String tagName, String start) {
        String m = escapeIdent(measurement);
        String n = escapeTag(tagName);
        String flux = String.format(
                "from(bucket: \"%s\") "
                        + "|> range(start: %s) "
                        + "|> filter(fn: (r) => r._measurement == \"%s\" and r.name == \"%s\" and r._field == \"printStart\" and r._value == 1) "
                        + "|> first()",
                bucket, start, m, n
        );

        List<FluxTable> tables = queryApi.query(flux);
        if (tables == null || tables.isEmpty()) return null;

        for (FluxTable table : tables) {
            for (FluxRecord rec : table.getRecords()) {
                if (rec.getTime() != null) {
                    return rec.getTime().toString();
                }
            }
        }
        return null;
    }

    // ========= 相对时间的“时间加权稼动率”（单 measurement）=========
    public Double queryAvailabilityWeightedRel(String measurement, String tagName, String range, boolean includeBusy) {
        String r = normalizeRange(range); // "-24h" 之类
        String includeBusyLiteral = includeBusy ? "true" : "false";
        String m = escapeIdent(measurement);
        String n = escapeTag(tagName);

        String flux = String.format("""
    import "experimental"

    // 计算相对时间窗口
    startDur = duration(v: "%s")
    tStop    = now()
    tStart   = experimental.addDuration(d: startDur, to: tStop)

    // 窗口内数据，仅保留时间和值
    data = from(bucket: "%s")
      |> range(start: tStart, stop: tStop)
      |> filter(fn: (r) => r._measurement == "%s" and r.name == "%s" and r._field == "statusCode")
      |> keep(columns: ["_time","_value"])

    // 起点锚点：取 tStart 前最后状态，时间改成 tStart，并保留同样的列
    init = from(bucket: "%s")
      |> range(start: -30d, stop: tStart)
      |> filter(fn: (r) => r._measurement == "%s" and r.name == "%s" and r._field == "statusCode")
      |> last()
      |> map(fn: (r) => ({ r with _time: tStart }))
      |> keep(columns: ["_time","_value"])

    // 合并（不做 distinct，后面 elapsed 再过滤 dur=0）
    base = union(tables: [init, data])
      |> sort(columns: ["_time"])

    // 结束锚点：补最后一段到 tStop，保持列一致
    endAnchor = base
      |> last()
      |> map(fn: (r) => ({ r with _time: tStop }))
      |> keep(columns: ["_time","_value"])

    // 加上结束锚点并计算每段持续秒数
    all = union(tables: [base, endAnchor])
      |> sort(columns: ["_time"])
      |> elapsed(unit: 1s, timeColumn: "_time", columnName: "dur")
      |> filter(fn: (r) => r.dur > 0)

    // includeBusy: 是否把 3 也计为运行
    includeBusy = %s

    ratio = all
      |> map(fn: (r) => ({
          r with
          run: if r._value == 1 or (includeBusy and r._value == 3) then float(v: r.dur) else 0.0,
          tot: float(v: r.dur)
      }))
      |> group(columns: [])
      |> reduce(
          fn: (r, accumulator) => ({ run: accumulator.run + r.run, tot: accumulator.tot + r.tot }),
          identity: { run: 0.0, tot: 0.0 }
      )
      |> map(fn: (r) => ({ _value: if r.tot > 0.0 then r.run / r.tot else 0.0 }))

    ratio
    """,
                r, bucket, m, n,
                bucket, m, n,
                includeBusyLiteral
        );

        List<FluxTable> tables = queryApi.query(flux);
        if (tables == null || tables.isEmpty()) return 0.0;
        for (FluxTable t : tables) {
            for (FluxRecord rec : t.getRecords()) {
                Object v = rec.getValue();
                if (v instanceof Number) return ((Number) v).doubleValue();
            }
        }
        return 0.0;
    }

    // 兜底：只允许 -24h/-12h/-7d 这类格式
    private String normalizeRange(String range) {
        if (range == null || range.isBlank()) return "-24h";
        String r = range.trim();
        if (!r.startsWith("-")) r = "-" + r;
        if (!r.matches("^-\\d+(ms|s|m|h|d|w)$")) return "-24h";
        return r;
    }

    // 构建 measurement 过滤条件: r._measurement == "m1" or r._measurement == "m2"
    private String buildMeasurementFilter(List<String> measurements) {
        if (measurements == null || measurements.isEmpty()) {
            return "false";
        }
        return measurements.stream()
                .filter(Objects::nonNull)
                .map(String::trim)
                .filter(s -> !s.isEmpty())
                .map(this::escapeIdent) // 统一使用转义
                .map(m -> "r._measurement == \"" + m + "\"")
                .collect(Collectors.joining(" or "));
    }

    private String sanitizeAggFn(String fn) {
        String f = (fn == null || fn.isBlank()) ? this.aggFn : fn.trim();
        Set<String> allowed = Set.of("mean", "median", "max", "min", "sum", "last", "first");
        return allowed.contains(f) ? f : "mean";
    }

    private String coalesce(String v, String fallback) {
        return (v == null || v.isBlank()) ? fallback : v;
    }

    private String buildFieldFilter(List<String> fields) {
        if (fields == null || fields.isEmpty()) {
            return "false";
        }
        return fields.stream()
                .filter(Objects::nonNull)
                .map(String::trim)
                .filter(s -> !s.isEmpty())
                .map(this::escapeTag)
                .map(f -> "r._field == \"" + f + "\"")
                .collect(Collectors.joining(" or "));
    }

    private String escapeTag(String v) {
        if (v == null) return "";
        return v.replace("\\", "\\\\").replace("\"", "\\\"");
    }

    private String escapeIdent(String v) {
        if (v == null) return "";
        return v.replace('-', '_').replace("\\", "\\\\").replace("\"", "\\\"");
    }

    // ========== 多 measurement：查询最新值（按字段分组后 last）==========
    public Optional<Map<String, Object>> queryLatestByNames(List<String> measurements, String tagName) {
        String mFilter = buildMeasurementFilter(measurements);
        String n = escapeTag(tagName);
        String flux = String.format("""
                        recent = from(bucket: "%s")
                          |> range(start: %s)
                          |> filter(fn: (r) => (%s) and r.name == "%s")
                          |> group(columns: ["_measurement","_field"])
                          |> last()
                          |> keep(columns: ["_measurement","_field","_value","_time"])
                        
                        statusFallback = from(bucket: "%s")
                          |> range(start: %s, stop: now())
                          |> filter(fn: (r) => (%s) and r.name == "%s" and (r._field == "statusCode"))
                          |> group(columns: ["_measurement","_field"])
                          |> last()
                          |> keep(columns: ["_measurement","_field","_value","_time"])
                        
                        union(tables: [recent, statusFallback])
                          |> group(columns: ["_measurement","_field"])
                          |> last()
                        """,
                bucket, latestRecentRange, mFilter, n,
                bucket, latestFallbackRange, mFilter, n
        );
        return queryLatestAsMap(flux);
    }

    // ========== 多 measurement：原始历史点 ==========
    public List<HistoryPoint> queryHistoryRawByNames(List<String> measurements, String tagName, List<String> fields, String start) {
        String mFilter = buildMeasurementFilter(measurements);
        String n = escapeTag(tagName);
        String fieldFilter = fields.stream()
                .map(f -> String.format("r._field == \"%s\"", escapeTag(f)))
                .collect(Collectors.joining(" or "));
        String flux = String.format(
                "from(bucket: \"%s\") "
                        + "|> range(start: %s) "
                        + "|> filter(fn: (r) => (%s) and r.name == \"%s\") "
                        + "|> filter(fn: (r) => %s) "
                        + "|> yield(name: \"raw\")",
                bucket, start, mFilter, n, fieldFilter
        );
        return queryPoints(flux);
    }

    // ========== 多 measurement：聚合历史点 ==========
    public List<HistoryPoint> queryHistoryAggByNames(List<String> measurements, String tagName, List<String> fields, String start, String agg) {
        String mFilter = buildMeasurementFilter(measurements);
        String n = escapeTag(tagName);
        String fieldFilter = fields.stream()
                .map(f -> String.format("r._field == \"%s\"", escapeTag(f)))
                .collect(Collectors.joining(" or "));
        String flux = String.format(
                "from(bucket: \"%s\") "
                        + "|> range(start: %s) "
                        + "|> filter(fn: (r) => (%s) and r.name == \"%s\") "
                        + "|> filter(fn: (r) => %s) "
                        + "|> aggregateWindow(every: %s, fn: mean, createEmpty: false) "
                        + "|> yield(name: \"agg\")",
                bucket, start, mFilter, n, fieldFilter, agg
        );
        return queryPoints(flux);
    }

    // ========== 多 measurement：首次 printStart=1 的时间 ==========
    public String queryFirstPrintStartTime(List<String> measurements, String tagName, String start) {
        String mFilter = buildMeasurementFilter(measurements);
        String n = escapeTag(tagName);
        String flux = String.format(
                "from(bucket: \"%s\") "
                        + "|> range(start: %s) "
                        + "|> filter(fn: (r) => (%s) and r.name == \"%s\" and r._field == \"printStart\" and r._value == 1) "
                        + "|> first()",
                bucket, start, mFilter, n
        );

        List<FluxTable> tables = queryApi.query(flux);
        if (tables == null || tables.isEmpty()) return null;
        for (FluxTable table : tables) {
            for (FluxRecord rec : table.getRecords()) {
                if (rec.getTime() != null) return rec.getTime().toString();
            }
        }
        return null;
    }

    // ========== 多 measurement：相对时间的“时间加权稼动率” ==========
    public Double queryAvailabilityWeightedRelMulti(List<String> measurements, String tagName, String range, boolean includeBusy) {
        String r = normalizeRange(range);
        String includeBusyLiteral = includeBusy ? "true" : "false";
        String mFilter = buildMeasurementFilter(measurements);
        String n = escapeTag(tagName);

        String flux = String.format("""
    import "experimental"

    startDur = duration(v: "%s")
    tStop    = now()
    tStart   = experimental.addDuration(d: startDur, to: tStop)

    data = from(bucket: "%s")
      |> range(start: tStart, stop: tStop)
      |> filter(fn: (r) => (%s) and r.name == "%s" and r._field == "statusCode")
      |> keep(columns: ["_time","_value"])

    init = from(bucket: "%s")
      |> range(start: -30d, stop: tStart)
      |> filter(fn: (r) => (%s) and r.name == "%s" and r._field == "statusCode")
      |> last()
      |> map(fn: (r) => ({ r with _time: tStart }))
      |> keep(columns: ["_time","_value"])

    base = union(tables: [init, data]) |> sort(columns: ["_time"])

    endAnchor = base |> last() |> map(fn: (r) => ({ r with _time: tStop })) |> keep(columns: ["_time","_value"])

    all = union(tables: [base, endAnchor])
      |> sort(columns: ["_time"])
      |> elapsed(unit: 1s, timeColumn: "_time", columnName: "dur")
      |> filter(fn: (r) => r.dur > 0)

    includeBusy = %s

    ratio = all
      |> map(fn: (r) => ({
          r with
          run: if r._value == 1 or (includeBusy and r._value == 3) then float(v: r.dur) else 0.0,
          tot: float(v: r.dur)
      }))
      |> group(columns: [])
      |> reduce(fn: (r, accumulator) => ({ run: accumulator.run + r.run, tot: accumulator.tot + r.tot }), identity: { run: 0.0, tot: 0.0 })
      |> map(fn: (r) => ({ _value: if r.tot > 0.0 then r.run / r.tot else 0.0 }))

    ratio
    """,
                r, bucket, mFilter, n,
                bucket, mFilter, n,
                includeBusyLiteral
        );

        List<FluxTable> tables = queryApi.query(flux);
        if (tables == null || tables.isEmpty()) return 0.0;
        for (FluxTable t : tables) {
            for (FluxRecord rec : t.getRecords()) {
                Object v = rec.getValue();
                if (v instanceof Number) return ((Number) v).doubleValue();
            }
        }
        return 0.0;
    }

    /**
     * 查询最新状态（带窗口聚合 + last 混合）
     * - aggFields: 对这些字段执行 aggregateWindow(every: window, fn: aggFn) 后取 last()
     * - lastFields: 对这些字段在近窗口(last)；若近窗口没有则在 fallback 窗口(last) 兜底
     * 结果：union 后对每个字段 group 再取最后一条
     */
    public Optional<Map<String, Object>> queryLatestAggByNames(List<String> measurements, String tagName, List<String> aggFields, String start,   // 为空则使用 latestRecentRange
                                                               String window,  // 为空则使用 aggWindow
                                                               String fn,      // 为空则使用 aggFn（默认 mean）
                                                               List<String> lastFields) {
        String mFilter = buildMeasurementFilter(measurements);
        String n = escapeTag(tagName);
        String aggFieldFilter = buildFieldFilter(aggFields);
        String lastFieldFilter = buildFieldFilter(lastFields);

        String startRange = coalesce(start, latestRecentRange);
        String windowDur = coalesce(window, aggWindow);
        String aggFunc = sanitizeAggFn(fn);

        String flux = String.format("""
                agg = from(bucket: "%s")
                  |> range(start: %s)
                  |> filter(fn: (r) => (%s) and r.name == "%s")
                  |> filter(fn: (r) => %s)
                  |> aggregateWindow(every: %s, fn: %s, createEmpty: false)
                  |> group(columns: ["_measurement","_field"])
                  |> last()
                  |> keep(columns: ["_measurement","_field","_value","_time"])
                
                recentRaw = from(bucket: "%s")
                  |> range(start: %s)
                  |> filter(fn: (r) => (%s) and r.name == "%s")
                  |> filter(fn: (r) => %s)
                  |> group(columns: ["_measurement","_field"])
                  |> last()
                  |> keep(columns: ["_measurement","_field","_value","_time"])
                
                fallbackRaw = from(bucket: "%s")
                  |> range(start: %s, stop: now())
                  |> filter(fn: (r) => (%s) and r.name == "%s")
                  |> filter(fn: (r) => %s)
                  |> group(columns: ["_measurement","_field"])
                  |> last()
                  |> keep(columns: ["_measurement","_field","_value","_time"])
                
                union(tables: [agg, recentRaw, fallbackRaw])
                  |> group(columns: ["_measurement","_field"])
                  |> last()
                """,
                bucket, startRange, mFilter, n, aggFieldFilter, windowDur, aggFunc,
                bucket, startRange, mFilter, n, lastFieldFilter,
                bucket, latestFallbackRange, mFilter, n, lastFieldFilter);

        return queryLatestAsMap(flux);
    }

    // ========== 多 measurement：仅聚合 ==========
    public Optional<Map<String, Object>> queryLatestAggOnlyByNames(
            List<String> measurements,
            String tagName,
            List<String> fields,
            String start,   // 为空用 latestRecentRange
            String window,  // 为空用 aggWindow
            String fn       // 为空用 aggFn（默认 mean）
    ) {
        String mFilter = buildMeasurementFilter(measurements);
        String n = escapeTag(tagName);
        String fieldFilter = buildFieldFilter(fields);
        String startRange = (start == null || start.isBlank()) ? latestRecentRange : start.trim();
        String windowDur = (window == null || window.isBlank()) ? aggWindow : window.trim();
        String aggFunc = sanitizeAggFn(fn);

        String flux = String.format("""
                from(bucket: "%s")
                  |> range(start: %s)
                  |> filter(fn: (r) => (%s) and r.name == "%s")
                  |> filter(fn: (r) => %s)
                  |> aggregateWindow(every: %s, fn: %s, createEmpty: false)
                  |> group(columns: ["_measurement","_field"])
                  |> last()
                  |> keep(columns: ["_measurement","_field","_value","_time"])
                """, bucket, startRange, mFilter, n, fieldFilter, windowDur, aggFunc);

        return queryLatestAsMap(flux);
    }

    // ========== 多 measurement：仅原始 last ==========
    public Optional<Map<String, Object>> queryLatestRawOnlyByNames(
            List<String> measurements,
            String tagName,
            List<String> fields,
            String recentStart, // 为空用 latestRecentRange
            String fallbackRange // 为空用 latestFallbackRange；传 "0s"/"off"/"none"/"disabled" 表示禁用回退
    ) {
        String mFilter = buildMeasurementFilter(measurements);
        String n = escapeTag(tagName);
        String fieldFilter = buildFieldFilter(fields);

        String recent = (recentStart == null || recentStart.isBlank()) ? latestRecentRange : recentStart.trim();
        String fallback = (fallbackRange == null || fallbackRange.isBlank()) ? latestFallbackRange : fallbackRange.trim();

        boolean useFallback = !isDisabledRange(fallback);

        String flux;
        if (useFallback) {
            flux = String.format("""
                        recent = from(bucket: "%s")
                          |> range(start: %s)
                          |> filter(fn: (r) => (%s) and r.name == "%s")
                          |> filter(fn: (r) => %s)
                          |> group(columns: ["_measurement","_field"])
                          |> last()
                          |> keep(columns: ["_measurement","_field","_value","_time"])
                        
                        fallback = from(bucket: "%s")
                          |> range(start: %s, stop: now())
                          |> filter(fn: (r) => (%s) and r.name == "%s")
                          |> filter(fn: (r) => %s)
                          |> group(columns: ["_measurement","_field"])
                          |> last()
                          |> keep(columns: ["_measurement","_field","_value","_time"])
                        
                        union(tables: [recent, fallback])
                          |> group(columns: ["_measurement","_field"])
                          |> last()
                    """, bucket, recent, mFilter, n, fieldFilter, bucket, fallback, mFilter, n, fieldFilter);
        } else {
            // 禁用回退：只查近窗口 recent
            flux = String.format("""
                    from(bucket: "%s")
                      |> range(start: %s)
                      |> filter(fn: (r) => (%s) and r.name == "%s")
                      |> filter(fn: (r) => %s)
                      |> group(columns: ["_measurement","_field"])
                      |> last()
                      |> keep(columns: ["_measurement","_field","_value","_time"])
                    """, bucket, recent, mFilter, n, fieldFilter);
        }

        return queryLatestAsMap(flux);
    }

    // 帮助方法：识别“禁用回退”的标记
    private boolean isDisabledRange(String r) {
        if (r == null) return false;
        String v = r.trim().toLowerCase(Locale.ROOT);
        return "0s".equals(v) || "off".equals(v) || "none".equals(v) || "disabled".equals(v);
    }


}